package com.zjl.Kafka.第03章_操作kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.HashMap;
import java.util.Map;

/**
 *  1.默认情况下，只要kafka有相关数据，就可以消费
 *           1).  生产者  -> kafka节点     查找事务管理器所在的节点
 *           2).  kafka主节点 -> 生产者    事务id =   事务值.hashCode % 50
 *           3).  生产者 -> 事务协调器      初始化生产者id ( __transaction_state-XX 这个文件的产生)
 *           4).  生产者 -> 事务协调器      将数据的分区信息传送到事务管理器
 *           5).  生产者  -> kafka节点     生产数据
 *           6).  kafka节点 -> 事务协调器   提交保存成功的数据分区信息
 *           7).  kafka节点 -> 生产者      返回单次数据是否成功
 *           8).  生产者 -> 事务协调器      结束事务
 *           9).  事务协调器 ->  事务状态   事务改成预提交状态
 *           10). 事务协调器-> kafka节点    给 Kafka 节点 说事务提交了
 *           11). 事务协调器 -> 事务状态    事务改成成功状态
 *
 *           9- 11 步骤中
 *                  事务预提交状态  之后 改成失败状态，这时数据保存了，但是状态为失败，
 *                  默认数据是可以消费的
 *  **************************************************************************************
 *  默认隔离级别：read_uncommitted  读未提交
 *          read_uncommitted   读未提交
 *          read_committed     读已提交
 *
 */
public class E2_kafka事务的隔离级别 {
    public static void main(String[] args) {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "82.157.71.243:9092");//连接地址
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//key 的 反序列化
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//value 的 反序列化
        config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");//提交了才可以读取

    }
}
